package dal

import (
	"context"
	"log"
	"pulsar_study/domain/entity"
	"pulsar_study/domain/event"
	"pulsar_study/infra"
	"sync"
	"time"

	"github.com/apache/pulsar-client-go/pulsar"
	"github.com/google/wire"
	"github.com/tidwall/gjson"
)

var EventPublisherSet = wire.NewSet(NewEventPublisher)

const topicPrefix = "persistent://public/default/"

type callbackFn = func(ctx context.Context, msg *entity.EventMessage) error
type EventPublisher struct {
	Trans        infra.GormTransClient
	EventDal     *DomainEventDal
	Client       pulsar.Client
	handlers     map[string][]callbackFn
	lock         sync.RWMutex
	producers    map[string]pulsar.Producer
	producerLock sync.RWMutex
}

var mqIns *EventPublisher
var mqOnce = sync.Once{}

func NewEventPublisher(trans infra.GormTransClient, eventDal *DomainEventDal) *EventPublisher {
	mqOnce.Do(func() {
		client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650", OperationTimeout: 30 * time.Second, ConnectionTimeout: 30 * time.Second})
		if err != nil {
			log.Fatal(err)
		}
		mqIns = &EventPublisher{
			Trans:     trans,
			EventDal:  eventDal,
			Client:    client,
			handlers:  make(map[string][]callbackFn),
			producers: make(map[string]pulsar.Producer),
		}
		go mqIns.initPulsar()
	})
	return mqIns
}

func (e *EventPublisher) getProducer(topic string) (pulsar.Producer, error) {
	e.producerLock.Lock()
	defer e.producerLock.Unlock()
	if producer, exists := e.producers[topic]; exists {
		return producer, nil
	}

	log.Printf("创建新的 Producer，Topic: %s, MaxPendingMessages: 1", topic)
	producer, err := e.Client.CreateProducer(pulsar.ProducerOptions{
		Topic:               topic,
		MaxPendingMessages:  1,
		SendTimeout:         3 * time.Second,
		DisableBatching:     true,
		BatchingMaxMessages: 1,
		CompressionType:     pulsar.LZ4,
	})
	if err != nil {
		return nil, err
	}
	e.producers[topic] = producer
	log.Printf("Producer 创建成功，Topic: %s", topic)
	return producer, nil
}

func (e *EventPublisher) AsyncPublish(ctx context.Context, ev event.DomainEvent) error {
	eventID, err := e.EventDal.Save(ctx, ev)
	if err != nil {
		log.Fatal("publish event error", err)
		return err
	}

	return e.Trans.Exec(ctx, func(ctx context.Context) error {
		// 更新发布状态
		err = e.EventDal.UpdateSendState(ctx, eventID, 1)
		if err != nil {
			log.Println("update send state error", err)
			return err
		}

		// 更新成功了才会继续发送消息
		producer, err := e.getProducer(e.getTopic(ev))
		if err != nil {
			log.Println("get producer error", err)
			return err
		}
		callback := func(msgID pulsar.MessageID, message *pulsar.ProducerMessage, err error) {
			if err != nil {
				// 发送失败的情况
				log.Printf("【失败】发送消息失败. Key: %s, Payload: %s, Error: %v\n",
					message.Key, string(message.Payload), err)
				// 在这里可以实现重试逻辑，例如将 producerMessage 放入一个重试 channel
			} else {
				// 发送成功的情况
				log.Printf("【成功】消息已确认. Message ID: %v, Key: %s\n",
					msgID, message.Key)
			}
		}

		log.Printf("准备发送异步消息，Key: %s, Topic: %s", ev.GetEntityID(), e.getTopic(ev))
		producer.SendAsync(ctx, &pulsar.ProducerMessage{
			Payload: []byte(ev.String()),
			Key:     ev.GetEntityID(),
		}, callback)
		if err != nil {
			log.Println("send message", err)
			return err
		}
		log.Printf("异步消息已提交，Key: %s", ev.GetEntityID())

		log.Printf("publish event id is %s", eventID)

		return nil
	})
}
func (e *EventPublisher) Publish(ctx context.Context, ev event.DomainEvent) error {
	eventID, err := e.EventDal.Save(ctx, ev)
	if err != nil {
		log.Fatal("publish event error", err)
		return err
	}

	return e.Trans.Exec(ctx, func(ctx context.Context) error {
		// 更新发布状态
		err = e.EventDal.UpdateSendState(ctx, eventID, 1)
		if err != nil {
			log.Println("update send state error", err)
			return err
		}

		// 更新成功了才会继续发送消息
		producer, err := e.getProducer(e.getTopic(ev))
		if err != nil {
			log.Println("get producer error", err)
			return err
		}
		_, err = producer.Send(ctx, &pulsar.ProducerMessage{
			Payload: []byte(ev.String()),
			Key:     ev.GetEntityID(),
		})
		if err != nil {
			log.Println("send message", err)
			return err
		}

		log.Printf("publish event id is %s", eventID)

		return nil
	})
}

func (e *EventPublisher) Subscribe(ctx context.Context, ev event.DomainEvent, fn callbackFn) *EventPublisher {
	e.lock.Lock()
	e.handlers[e.getTopic(ev)] = append(e.handlers[e.getTopic(ev)], fn)
	defer e.lock.Unlock()
	return e
}

func (e *EventPublisher) getTopic(ev event.DomainEvent) string {
	return topicPrefix + ev.GetEventName()
}

func (e *EventPublisher) initPulsar() {
	ctx := context.Background()
	// 初始化消费者
	consumer, err := e.Client.Subscribe(pulsar.ConsumerOptions{
		TopicsPattern:    topicPrefix + "*",
		SubscriptionName: "pulsar-study",
		Type:             pulsar.Shared,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer consumer.Close()
	worker := &MqWorker{
		ch: make(chan pulsar.Message, 32),
	}
	go worker.start(ctx, consumer, e)
	// 定义处理消息的逻辑
	for {
		message, err := consumer.Receive(ctx)
		if err != nil {
			log.Fatal(err)
		}
		worker.ch <- message
	}
}

func (e *EventPublisher) CloseProducers() {
	e.producerLock.Lock()
	defer e.producerLock.Unlock()

	for topic, producer := range e.producers {
		producer.Close()
		delete(e.producers, topic)
		log.Printf("Closed producer for topic: %s", topic)
	}
}

func (e *EventPublisher) processMessage(ctx context.Context, consumer pulsar.Consumer, msg pulsar.Message) {
	// 重复消费检测
	eventID := gjson.GetBytes(msg.Payload(), "event_id").String()
	if eventID == "" {
		eventID = msg.ID().String()
	}
	unprocessed, err := e.EventDal.MarkEventProcessed(ctx, msg.Topic(), eventID)
	if err == nil && !unprocessed {
		log.Printf("eventID:%s 重复消费！", eventID)
		_ = consumer.Ack(msg)
		return
	}

	// 消费消息
	fns := e.handlers[msg.Topic()]
	if len(fns) == 0 {
		consumer.Nack(msg)
		log.Println("no subscriber")
	}
	for _, callFn := range fns {
		if err := callFn(ctx, &entity.EventMessage{
			ID:      msg.Key(),
			Content: string(msg.Payload()),
			Topic:   msg.Topic(),
		}); err != nil {
			log.Printf("error:%s, nack message, topic:%s, eventID:%s", err.Error(), msg.Topic(), eventID)
			consumer.Nack(msg)
			//_ = consumer.Ack(msg)
			break
		} else {
			_ = consumer.Ack(msg)
			log.Println("ack msg, process success")
		}
	}
}

type MqWorker struct {
	ch chan pulsar.Message
}

func (w *MqWorker) start(ctx context.Context, consumer pulsar.Consumer, e *EventPublisher) {
	for {
		select {
		case <-ctx.Done():
			return
		case msg, ok := <-w.ch:
			if !ok {
				return
			}
			e.processMessage(ctx, consumer, msg)
		}
	}
}
